package com.infinova.datax.apireader;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.StopWatch;
import cn.hutool.core.map.MapUtil;
import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONPath;
import com.infinova.datax.common.http.HttpClient;
import com.infinova.datax.common.http.HttpRequest;
import com.infinova.datax.common.http.OkHttp3Client;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

import static com.alibaba.fastjson2.JSONWriter.Feature.PrettyFormat;
import static com.infinova.datax.apireader.Config.PageType.PAGE;

/**
 * @author bigbird-0101
 * @date 2023-07-24 11:18
 */
public class ApiReader extends Reader {
    public static class Job extends Reader.Job{
        private final static Logger LOGGER =LoggerFactory.getLogger("com.infinova.datax.apireader.ApiReader.Job");
        Configuration pluginJobConf=null;
        @Override
        public List<Configuration> split(int i) {
            Config convert = Config.convert(pluginJobConf);
            if(convert.getEnablePage()){
                long page= convert.getPageParam().map(s->Long.valueOf(String.valueOf(s.getValue()))).orElse(0L);
                int pageSize=convert.getPageSizeParam().map(s->Integer.valueOf(String.valueOf(s.getValue()))).orElse(100);
                Integer maxLoopThreshold = pluginJobConf.getInt("maxLoopThreshold", 50000);
                List<Segment> split = Segment.split(maxLoopThreshold, i, page, PAGE.equals(convert.getPageType())?1:pageSize);
                LOGGER.info("segment {}",JSONObject.toJSONString(split,PrettyFormat));
                List<Configuration> result=new ArrayList<>();
                for(Segment segment:split){
                    Configuration clone = pluginJobConf.clone();
                    clone.set("segment",segment);
                    result.add(clone);
                }
                return result;
            }else {
                return Collections.singletonList(pluginJobConf);
            }
        }

        @Override
        public void init() {
            pluginJobConf = super.getPluginJobConf();
        }

        @Override
        public void destroy() {

        }
    }

    public static class Task extends Reader.Task{
        private final static Logger LOGGER =LoggerFactory.getLogger("com.infinova.datax.apireader.ApiReader.Task");
        private HttpClient httpClient;

        private Config config;

        @Override
        public void startRead(RecordSender recordSender) {
            StopWatch stopWatch=new StopWatch("startRead");
            stopWatch.start("convertIterator");
            Iterator<HttpRequest> httpRequestIterator = config.convertIterator();
            stopWatch.stop();
            while(httpRequestIterator.hasNext()){
                HttpRequest request = httpRequestIterator.next();
                if(null==request){
                    continue;
                }
                stopWatch.start("send Request ");
                JSONObject result = httpClient.handle(request, JSONObject.class);
                if(MapUtil.isEmpty(result)){
                    break;
                }
                stopWatch.stop();
                stopWatch.start("get real result ");
                Object realResult= result;
                String resultDataPath = config.getResultDataPath();
                if(null!=resultDataPath){
                    realResult= JSONPath.eval(result,resultDataPath);
                }
                if(LOGGER.isInfoEnabled()){
                    LOGGER.info("realResult {}",JSONObject.toJSONString(realResult,PrettyFormat));
                }
                stopWatch.stop();

                stopWatch.start("do send record ");
                if(realResult instanceof JSONObject){
                    Record record = recordSender.createRecord();
                    JSONObject jsonObject = (JSONObject) realResult;
                    if(MapUtil.isEmpty(jsonObject)){
                        break;
                    }
                    //从返回的数据当中 拉去配置的数据模型数据
                    addRecordColumn(record, jsonObject);
                    recordSender.sendToWriter(record);
                }else if(realResult instanceof JSONArray){
                    JSONArray jsonArray = (JSONArray) realResult;
                    if(CollUtil.isEmpty(jsonArray)){
                        break;
                    }
                    for(Object object: jsonArray) {
                        JSONObject jsonObject= (JSONObject) object;
                        Record record = recordSender.createRecord();
                        addRecordColumn(record, jsonObject);
                        recordSender.sendToWriter(record);
                    }
                }
                stopWatch.stop();
                try {
                    //避免请求太频繁
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            if(LOGGER.isInfoEnabled()){
                LOGGER.info("执行耗时结果{}",stopWatch.prettyPrint());
            }
        }

        private void addRecordColumn(Record record, JSONObject jsonObject) {
            jsonObject.forEach((k, v)-> config.getDataModels()
                    .stream()
                    .filter(s->s.getKey().equals(k))
                    .findFirst()
                    .ifPresent(s->{
                        Column column = s.getValueType().convertTarget(v);
                        record.addColumn(column);
                    }));
        }

        @Override
        public void init() {
            this.config = Config.convert(super.getPluginJobConf());
            this.httpClient=new OkHttp3Client();
        }

        @Override
        public void destroy() {
            this.httpClient.destroy();
        }
    }
}
